#!/bin/bash
#设置任务异常退出
set -eo pipefail
dt='{{ execution_date | cst_ds }}'
name='末端时效退转明细'
# 在退出时清理文件
trap "rm -f /tmp/${name}_${dt}.zip /tmp/${name}_${dt}.csv && echo 'removed tmp files'" EXIT
# 创建txt文件
touch /tmp/${name}_${dt}.txt
#将中文字段名输入文件顶部
#将查询结果追加到文件中
{
  echo -e "运单号\t来源名字\t末端中心发件时间\t规划签收网点编码\t规划签收网点名称\t规划签收时间\t派件派仓班次\t时效签收时间\t实效签收网点编码\t实效签收网点名称\t退回转寄件申请时间\t时效签收是否准点\t">/tmp/${name}_${dt}.txt;
  spark-sql --driver-memory 4G --num-executors 10  --executor-memory 8g --executor-cores 6  --conf spark.executor.memoryOverhead=4g --conf spark.driver.maxResultSize=3g -S -e "
  with t_aging_sign as (--时效签收
    select billcode,
           case when length(min(wid_columns)[0]) = 0 then null else min(wid_columns)[0] end fst_aging_sign_time,
           case when length(min(wid_columns)[1]) = 0 then null else min(wid_columns)[1] end fst_aging_sign_network_code,
           case when length(min(wid_columns)[2]) = 0 then null else min(wid_columns)[2] end fst_aging_sign_scan_type,
           case when length(min(wid_columns)[3]) = 0 then null else min(wid_columns)[3] end fst_aging_sign_user_code,
           case when length(min(wid_columns)[4]) = 0 then null else min(wid_columns)[4] end fst_aging_sign_pistolid,--巴枪id PDD开头是桃花岛其他是自助对接
           case when length(max(wid_columns)[0]) = 0 then null else max(wid_columns)[0] end lst_aging_sign_time,
           case when length(max(wid_columns)[1]) = 0 then null else max(wid_columns)[1] end lst_aging_sign_network_code,
           case when length(max(wid_columns)[2]) = 0 then null else max(wid_columns)[2] end lst_aging_sign_scan_type,
           case when length(max(wid_columns)[3]) = 0 then null else max(wid_columns)[3] end lst_aging_sign_user_code,
           case when length(max(wid_columns)[4]) = 0 then null else max(wid_columns)[4] end lst_aging_sign_pistolid
    from (
             select billcode,
                    split(concat(scantime, '\001',
                                 coalesce(scansitecode, ''), '\001',
                                 coalesce(scantype, ''), '\001',
                                 coalesce(send_deliver_usercode, ''), '\001',
                                 coalesce(pistolid, '')
                              ), '\001') as wid_columns
             from (
                      select billcode,
                             delivertime as scantime,
                             scansitecode,
                             '快件签收'      as scantype,
                             send_deliver_usercode,
                             null        as pistolid
                      from jms_ods.tab_barscan_sign
                      where dt between date_format(date_add('{{ execution_date |  cst_ds }}', -15), 'yyyy-MM-dd')
                          and date_format(date_add('{{ execution_date |  cst_ds }}', 0), 'yyyy-MM-dd')
                        and length(billcode) = 15
                      union all
                      select billcode,
                             scantime,
                             scansitecode,
                             scantype,
                             send_deliver_usercode,
                             pistolid
                      from jms_ods.tab_barscan_collect
                      where dt between date_format(date_add('{{ execution_date |  cst_ds }}', -15), 'yyyy-MM-dd')
                          and date_format(date_add('{{ execution_date |  cst_ds }}', 0), 'yyyy-MM-dd')
                        and length(billcode) = 15
                        and scantype in ('代理点收入扫描', '出库扫描', '入库扫描')
                  ) t
         ) t
    group by billcode
)
select
              a.billcode , --运单号
              order_source_name,--订单来源原始名字
              end_center_send_time,--末端中心/集散发件时间
              final_sign_network_code,--t规划签收网点编码
              final_sign_network_name,--t规划签收网点名称
              final_plan_sign_time,  --t规划签收时间
              deliver_shift, --t派件派仓班次
              c.fst_aging_sign_time as aging_sign_time, --t时效签收时间
              c.fst_aging_sign_network_code as  aging_sign_network_code,--t实效签收网点编码
              d.name as aging_sign_network_name,--t实效签收网点名称
              reback_transfer_apply_time,--t退回转寄件申请时间
              if(c.fst_aging_sign_time<final_plan_sign_time,1,0)  as is_age  --t时效签收是否准点
          from ( select
          billcode ,
              order_source_name,
              end_agent_name ,
              end_provider_name,
              end_center_send_time,
              send_next_station_code,
              send_next_station_name,
              final_sign_network_code,
              final_sign_network_name,
              final_plan_sign_time,
              deliver_shift,
              end_network_arrival_time,
              deliver_time,
              aging_sign_time, --时效签收时间
              aging_sign_scan_type,
              aging_sign_network_code,--时效签收网点编码
              aging_sign_network_name,--时效签收网点名称
              actual_sign_network_code,
              actual_sign_network_name,
              actual_sign_time,  --实际签收时间
              actual_sign_scan_type,
              is_intercept,
              intercept_registration_time,
              is_reback_transfer,
              reback_transfer_apply_time,
              is_end_piece,
              end_time,
              deadline_signing_time
          from jms_dwd.dwd_terminal_sign_all_detail_dt
          where dt>=date_sub('{{ execution_date |  cst_ds }}',3)
          and dt<='{{ execution_date |  cst_ds }}'
          and is_reback_transfer=1
          and TO_DATE(final_plan_sign_time)='{{ execution_date |  cst_ds }}'
          )a
          left semi JOIN (
          SELECT waybill_no,
          apply_type_name,
          print_network_code,
          print_network_name,
          apply_time,
          print_time
          from jms_ods.tab_reback_transfer_express
          where dt='{{ execution_date |  cst_ds }}'
          and  apply_type_code=4
          and origin_network_flag=1
          )b
          on a.billcode=b.waybill_no
          left join t_aging_sign c
          on a.billcode=c.billcode
  left join (select code,
            name,
            agent_id,
            agent_code,
            agent_name,
            franchisee_code,
            franchisee_name,
            provider_id,
            provider_desc,
            city_id,
            city_desc,
            area_id,
            area_desc,
            type_id,
            case when type_id = 335 then 1 when type_id = 336 and is_entrepot = 1 then 2 else 3 end station_type--1 中心 2 集散 3 网点
      from (
             select t_network.*,
                    row_number() over (partition by t_network.code order by t_network.update_time desc) bill_rank
             from jms_dim.dim_sys_network_detail_dt t_network
             where dt = date_format(date_add('{{ execution_date |  cst_ds }}', 0), 'yyyy-MM-dd')
            ) t where bill_rank = 1)d
          on c.fst_aging_sign_network_code=d.code;

" >>/tmp/${name}_${dt}.txt
}
# 替换分隔符，并输出csv
cat /tmp/${name}_${dt}.txt | tr -s '\t' ',' > /tmp/${name}_${dt}.csv
# 压缩并删除 csv 文件
zip -jrm /tmp/${name}_${dt}.zip /tmp/${name}_${dt}.csv
# 上传文件
/usr/local/python3/bin/python3  /usr/local/airflow/dags/utils/operators/upload_oss.py ${name}/${name}_${dt}.zip /tmp/${name}_${dt}.zip
# zip 文件将在进程退出后被自动清理

#声明文件大小
data_size=$(ls -lh /tmp/${name}_${dt}.zip | awk '{print $5}')
#打印文件信息
Message="${name}_${dt}.zip 上传成功 大小: $data_size"
#配置钉钉告警
curl 'https://oapi.dingtalk.com/robot/send?access_token=816a8f35fd84432ffa6bff9d7714abc322628bb4a8ded86908550d2335aba613' \
   -H 'Content-Type: application/json' \
   -d "
  {\"msgtype\": \"text\",
    \"text\": {
        \"content\": \"监控内容:$Message\"
     }
  }" || true
rm -f  /tmp/${name}_${dt}.txt